#include "wl_transport.h"

#include "wl_mutex.h"
#include "wl_trace.h"

#include <arpa/inet.h>
#include <dirent.h>
#include <errno.h>
#include <netdb.h>
#include <netinet/in.h>
#include <netinet/tcp.h>
#include <poll.h>
#include <pthread.h>
#include <sys/epoll.h>
#include <sys/ioctl.h>
#include <sys/prctl.h>
#include <sys/select.h>
#include <sys/socket.h>
#include <sys/types.h>
#include <time.h>
#include <unistd.h>
#include <ak_thread.h>

#ifdef __cplusplus
extern "C" {
#endif /* __cplusplus */

typedef struct __wl_transport_message {
    WL_SOCKET socket;
    wl_int32  event;
} wl_transport_message;

#define MAX_SOCK_NUM 16

#define START_OF_SOCKFD 1

#define EPOLL_EVENT_NUM 10

#define CHECK_SOCKET(socket) (START_OF_SOCKFD > socket || MAX_SOCK_NUM < socket) ? wl_false : wl_true

typedef struct __wl_transport {
    wl_bool              inuse;
    WL_SOCKET_DESCRIPTOR fd;
    wl_int32             taskid;
    wl_int32             event;
    wl_int32             type;
    wl_sock_callback     callback;
    void*                context;
} wl_transport;

typedef struct {
    wl_bool      initialized;
    wl_bool      exit;
    wl_int32     count;
    wl_transport transports[MAX_SOCK_NUM];
    WL_MUTEX     mutex;
    ak_pthread_t thread;
    wl_int32     epollfd;
} wl_transport_global;

wl_transport_global s_transport_global;

static WL_SOCK_EVENT translate_transport_event_to_socket_event(wl_int32 transport_event);
static wl_int32      set_socket_select_event(wl_int32 transport_index, wl_transport* transport);

static WL_SOCK_EVENT translate_transport_event_to_socket_event(wl_int32 transport_event) {
    switch (transport_event) {
        case WL_TRANSPORT_EVENT_READ:
            return WL_SOCK_READ_AVAILABLE;
        case WL_TRANSPORT_EVENT_WRITE:
            return WL_SOCK_WRITE_AVAILABLE;
        case WL_TRANSPORT_EVENT_ACCEPT:
            return WL_SOCK_NEW_CONNECTION;
        case WL_TRANSPORT_EVENT_CONNECT:
            return WL_SOCK_CONNECTION_SUCCESS;
        case WL_TRANSPORT_EVENT_CLOSE:
            return WL_SOCK_CONNECTION_LOST;
        case WL_TRANSPORT_EVENT_LOST:
            return WL_SOCK_CONNECTION_LOST;
        case WL_TRANSPORT_EVENT_EXCEPTION:
            return WL_SOCK_CONNECTION_FAILED;
        default:
            break;
    }
    return WL_SOCK_ERROR_UNKNOWN;
}

static wl_int32 set_socket_select_event(wl_int32 transport_index, wl_transport* transport) {
    struct epoll_event ev;

    ev.events  = EPOLLET | EPOLLIN | EPOLLHUP | EPOLLERR;
    ev.data.fd = transport_index;

    if (transport->event & WL_TRANSPORT_EVENT_WRITE || transport->event & WL_TRANSPORT_EVENT_CONNECT) {
        ev.events |= EPOLLOUT;
    }

    if (0 > epoll_ctl(s_transport_global.epollfd, EPOLL_CTL_DEL, transport->fd, wl_null)) {
        WLLOGW("epoll_ctl EPOLL_CTL_DEL socket %d fail", transport->fd);
    }

    if (0 > epoll_ctl(s_transport_global.epollfd, EPOLL_CTL_ADD, transport->fd, &ev)) {
        WLLOGW("epoll_ctl socket %d fail", transport->fd);
        return -1;
    }

    // WLLOGT("epoll_ctl sd %d ev %x event %d", transport->fd, ev.events, transport->event);
    return 0;
}

void* transport_main_thread(void* param) {
    wl_transport* transport = wl_null;
    struct epoll_event events[EPOLL_EVENT_NUM];
    wl_int32 nfds = 0;
    wl_int32 i = 0;
    WL_TRANSPORT_EVENT transport_event;

    prctl(PR_SET_NAME, "transport_main_thread");

    UNUSED(param);
    s_transport_global.epollfd = epoll_create(MAX_SOCK_NUM);
    WLLOGT("transport start EPOLLIN %d EPOLLOUT %d EPOLLHUP %d EPOLLERR %d", EPOLLIN, EPOLLOUT, EPOLLHUP, EPOLLERR);
    while (!s_transport_global.exit) {
        nfds = epoll_wait(s_transport_global.epollfd, events, EPOLL_EVENT_NUM, -1);
        for (i = 0; i < nfds; i++) {
            wl_mutex_acquire(&s_transport_global.mutex);
            transport_event = WL_TRANSPORT_EVENT_NONE;
            transport = &s_transport_global.transports[events[i].data.fd];
            WLLOGT("socket %d events %d transport event %x",
                   events[i].data.fd + START_OF_SOCKFD,
                   events[i].events,
                   transport->event);
            if (EPOLLIN & events[i].events) {
                if (transport->event & WL_TRANSPORT_EVENT_ACCEPT) {
                    transport_event = WL_TRANSPORT_EVENT_ACCEPT;
                    transport->event &= ~WL_TRANSPORT_EVENT_ACCEPT;
                } else {
                    transport_event = WL_TRANSPORT_EVENT_READ;
                }
            } else if (EPOLLOUT & events[i].events) {
                //连接成功,添加可读事件监听
                if (transport->event & WL_TRANSPORT_EVENT_CONNECT) {
                    struct epoll_event ev;

                    transport_event = WL_TRANSPORT_EVENT_CONNECT;
                    transport->event &= ~WL_TRANSPORT_EVENT_CONNECT;

                    ev.events = EPOLLET | EPOLLIN | EPOLLHUP | EPOLLERR;
                    ev.data.fd = events[i].data.fd;
                    epoll_ctl(s_transport_global.epollfd, EPOLL_CTL_DEL, transport->fd, wl_null);
                    if (0 > epoll_ctl(s_transport_global.epollfd, EPOLL_CTL_ADD, transport->fd, &ev)) {
                        WLLOGW("epoll_ctl sd %d EPOLL_CTL_ADD fd %d fail", s_transport_global.epollfd, transport->fd);
                        transport_event = WL_TRANSPORT_EVENT_EXCEPTION;
                    }
                } else {
                    transport_event = WL_TRANSPORT_EVENT_WRITE;
                }
            } else if ((EPOLLHUP & events[i].events) || (EPOLLERR & events[i].events)) {
                if (transport->event & WL_TRANSPORT_EVENT_CONNECT) {
                    transport_event = WL_TRANSPORT_EVENT_LOST;
                    transport->event &= ~WL_TRANSPORT_EVENT_CONNECT;
                } else {
                    transport_event = WL_TRANSPORT_EVENT_CLOSE;
                }
                epoll_ctl(s_transport_global.epollfd, EPOLL_CTL_DEL, transport->fd, wl_null);
            } else {
                WLLOGT("invalid events %d", events[i].events);
            }
            wl_mutex_release(&s_transport_global.mutex);

            if (transport_event) {
                wl_transport_message transport_message;
                wl_task_message message;
                memset(&transport_message, 0, sizeof(wl_transport_message));
                memset(&message, 0, sizeof(wl_task_message));
                transport_message.event = transport_event;
                transport_message.socket = events[i].data.fd + START_OF_SOCKFD;
                message.type = WL_TASK_MSG_SOCKET;
                memcpy(&message.data, &transport_message, sizeof(wl_transport_message));
                message.datalen = sizeof(wl_transport_message);
                wl_send_task_message(transport->taskid, &message);
                // WLLOGT(
                //     "send taskid %d socket %d event %d", transport->taskid, transport_message.socket,
                //     transport_event);
            }
        }
    }

    close(s_transport_global.epollfd);
    ak_thread_exit();
    WLLOGT("transport exit");
    return wl_null;
}

wl_bool wl_transport_set_socket_event(WL_SOCKET socket, wl_int32 event) {
    if (CHECK_SOCKET(socket)) {
        wl_transport* transport = wl_null;
        wl_mutex_acquire(&s_transport_global.mutex);
        transport = &s_transport_global.transports[socket - START_OF_SOCKFD];
        if (transport->inuse && WL_INVALID_SOCKET != transport->fd) {
            transport->event = event;
            set_socket_select_event(socket - START_OF_SOCKFD, transport);
        }
        wl_mutex_release(&s_transport_global.mutex);
        return wl_true;
    } else {
        return wl_false;
    }
}

WL_SOCKET_DESCRIPTOR wl_transport_get_socket_descriptor(WL_SOCKET socket) {
    WL_SOCKET_DESCRIPTOR sd = WL_INVALID_SOCKET;
    if (CHECK_SOCKET(socket)) {
        wl_transport* transport = wl_null;
        wl_mutex_acquire(&s_transport_global.mutex);
        transport = &s_transport_global.transports[socket - START_OF_SOCKFD];
        if (transport->inuse && WL_INVALID_SOCKET != transport->fd) {
            sd = transport->fd;
        }
        wl_mutex_release(&s_transport_global.mutex);
    }
    return sd;
}

wl_int32 wl_transport_get_socket_taskid(WL_SOCKET socket) {
    if (CHECK_SOCKET(socket)) {
        return s_transport_global.transports[socket - START_OF_SOCKFD].taskid;
    }
    return -1;
}

wl_int32 wl_transport_set_socket_callback(WL_SOCKET socket, wl_sock_callback callback, void* context) {
    if (CHECK_SOCKET(socket)) {
        wl_transport* transport = wl_null;
        wl_mutex_acquire(&s_transport_global.mutex);
        transport = &s_transport_global.transports[socket - START_OF_SOCKFD];
        // WLLOGT("socket %d callback %p context %p inuse %d fd %d",
        //        socket,
        //        callback,
        //        context,
        //        transport->inuse,
        //        transport->fd);
        if (transport->inuse && WL_INVALID_SOCKET != transport->fd) {
            transport->callback = callback;
            transport->context  = context;
        }
        wl_mutex_release(&s_transport_global.mutex);
        return 0;
    } else {
        WLLOGW("socket %d invalid", socket);
        return -1;
    }
}

wl_int32 wl_transport_init(void) {
    WLLOGW("wl_transport_init");
    if (!s_transport_global.initialized) {
        s_transport_global.exit = wl_false;

        if (0 != wl_mutex_create("wlsdk_transport", &s_transport_global.mutex)) {
            WLLOGW("wl_mutex_create fail");
            return -1;
        }
        //创建线程
        if (0 != ak_thread_create(&s_transport_global.thread, transport_main_thread, wl_null, ANYKA_THREAD_MIN_STACK_SIZE, -1)) {
            wl_mutex_delete(&s_transport_global.mutex);
            WLLOGW("pthread_create fail");
            return -1;
        }

        ak_thread_detach(s_transport_global.thread);

        s_transport_global.initialized = wl_true;
        WLLOGT("wl_transport_init success");
    }
    return 0;
}

void wl_transport_end(void) {
    if (s_transport_global.initialized) {
        wl_int32 index           = 0;
        s_transport_global.exit  = wl_true;
        s_transport_global.count = 0;
        wl_mutex_acquire(&s_transport_global.mutex);

        for (index = 0; index < MAX_SOCK_NUM; index++) {
            if (s_transport_global.transports[index].inuse) {
                close(s_transport_global.transports[index].fd);
                memset(&s_transport_global.transports[index], 0, sizeof(wl_transport));
            }
        }

        wl_mutex_release(&s_transport_global.mutex);
        ak_thread_join(s_transport_global.thread);
        wl_mutex_delete(&s_transport_global.mutex);
        s_transport_global.initialized = wl_false;
    }
}

WL_SOCKET wl_transport_add_socket(WL_SOCKET_DESCRIPTOR sd, wl_int32 type, wl_int32 taskid) {
    wl_int32 index = 0;

    wl_mutex_acquire(&s_transport_global.mutex);
    for (index = 0; index < MAX_SOCK_NUM; index++) {
        if (!s_transport_global.transports[index].inuse) {
            WL_SOCKET     socket    = index + START_OF_SOCKFD;
            wl_transport* transport = &s_transport_global.transports[index];
            memset(transport, 0, sizeof(wl_transport));
            transport->fd     = sd;
            transport->taskid = taskid;
            transport->inuse  = wl_true;
            transport->type   = type;
            transport->event  = WL_TRANSPORT_EVENT_NONE;

            s_transport_global.count++;
            wl_mutex_release(&s_transport_global.mutex);
            return socket;
        }
    }
    wl_mutex_release(&s_transport_global.mutex);
    WLLOGW("add task %d sd %d fail", sd, taskid);
    return WL_INVALID_SOCKET;
}

wl_int32 wl_transport_delete_socket(WL_SOCKET socket) {
    if (CHECK_SOCKET(socket)) {
        wl_transport* transport = wl_null;
        wl_mutex_acquire(&s_transport_global.mutex);
        transport = &s_transport_global.transports[socket - START_OF_SOCKFD];
        if (transport->inuse && WL_INVALID_SOCKET != transport->fd) {
            if (0 > epoll_ctl(s_transport_global.epollfd, EPOLL_CTL_DEL, transport->fd, wl_null)) {
                WLLOGW("epoll_ctl EPOLL_CTL_DEL socket %d fail", transport->fd);
            }
            s_transport_global.count--;
            transport->inuse    = wl_false;
            transport->taskid   = 0;
            transport->fd       = WL_INVALID_SOCKET;
            transport->callback = wl_null;
            transport->context  = wl_null;
        }
        wl_mutex_release(&s_transport_global.mutex);
        return 0;
    } else {
        return -1;
    }
}

wl_int32 wl_transport_process_message(wl_task_message* message) {
    wl_transport_message transport_message;
    wl_sock_callback     callback = wl_null;
    void*                context  = wl_null;
    if (wl_null == message || sizeof(wl_transport_message) != message->datalen) {
        WLLOGW("message is null");
        return -1;
    }
    memcpy(&transport_message, &(message->data), message->datalen);
    if (CHECK_SOCKET(transport_message.socket)) {
        wl_transport* transport = wl_null;
        wl_mutex_acquire(&s_transport_global.mutex);
        transport = &s_transport_global.transports[transport_message.socket - START_OF_SOCKFD];
        callback  = transport->callback;
        context   = transport->context;
        wl_mutex_release(&s_transport_global.mutex);

        if (callback) {
            WL_SOCK_EVENT socket_event = translate_transport_event_to_socket_event(transport_message.event);
            WLLOGT("socket: %d event: %d socket_event %d",
                   transport_message.socket,
                   transport_message.event,
                   socket_event);
            callback(transport_message.socket, socket_event, context);
        } else {
            WLLOGW("socket: %d callback is null", transport_message.socket);
        }
    } else {
        WLLOGW("invalid socket %d", transport_message.socket);
    }

    return 0;
}

#ifdef __cplusplus
}
#endif /* __cplusplus */
